package com.etc;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;

public class ReduceSideJoin {
    public static class ReduceSideJoinMapper extends Mapper<LongWritable,Text, Text,JoinBean>{
        String fileName = null;
        JoinBean bean = new JoinBean();
        Text K = new Text();
        @Override
        protected void map(LongWritable key, Text value,Mapper <LongWritable,Text,Text,JoinBean>.Context context) throws IOException, InterruptedException {
            String[]  fields = value.toString().split(",");
            if(fileName.startsWith("order")){
                  bean.set(fields[0],fields[1],"NULL",-1,"NULL","order");
            }else{
                  bean.set("NULL",fields[0],fields[1],Integer.getInteger(fields[2]),fields[3],"user");
            }
          K.set(bean.getUserId());
            context.write(K,bean);
        }
    }
    public static class ReduceSideJoinReducer extends Reducer<Text,JoinBean,JoinBean, NullWritable> {

        @Override
        protected void reduce(Text key, Iterable<JoinBean> beans, Reducer<Text,JoinBean,JoinBean,NullWritable>.Context context) throws IOException, InterruptedException {
            ArrayList<JoinBean> orderList = new ArrayList<JoinBean>();
            JoinBean userBean = null;

            try {
                     for (JoinBean bean : beans) {
                           if ("order".equals(userBean.getTableName())) {
                            JoinBean newBean = new JoinBean();
                            BeanUtils.copyProperties(newBean, bean);
                            orderList.add(newBean);
                    } else {
                               userBean = new JoinBean();
                               BeanUtils.copyProperties(userBean, bean);
                           }
                        }
                for (JoinBean bean : orderList) {
                    bean.setTableName(userBean.getUserName());
                    bean.setUserAge(userBean.getUserAge());
                    bean.setUserFriend(userBean.getUserFriend());

                    context.write(bean, NullWritable.get());
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInt("order.top.n", 2);

        Job job = Job.getInstance(conf);

        job.setJarByClass(ReduceSideJoin.class);

        job.setMapperClass(ReduceSideJoinMapper.class);
        job.setReducerClass(ReduceSideJoinReducer.class);

        job.setNumReduceTasks(2);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(JoinBean.class);

        job.setOutputKeyClass(JoinBean.class);
        job.setOutputValueClass(NullWritable.class);

        File file = new File("f:all/a/join1/a");
        if(file.exists()){
            FileUtils.deleteDirectory(file);
        }

        FileInputFormat.setInputPaths(job, new Path("f:/all/a/join1"));
        FileOutputFormat.setOutputPath(job, new Path("f:all/a/join1/a"));

        job.waitForCompletion(true);
    }
}
